package com.ctrip.ibu.member.client.consumer;


import com.ctrip.ibu.member.client.common.BrokerStatusUtil;
import com.ctrip.ibu.member.client.module.BrokerInfo;
import com.ctrip.ibu.member.client.module.MessageInfo;
import com.ctrip.ibu.member.common.util.NumberUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Instant;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;

/**
 * 消费者client, 开启关闭消费者
 *
 * @author he.shaorong
 * @create 2023/3/27 16:21
 * @since 1.0.0
 */
public class ConsumerClient {

    private static final Logger logger = LoggerFactory.getLogger(ConsumerClient.class);

    /**
     * 创建任务队列   10 为线程数量
     */
    private static ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(10);

    /**
     * 缓存正在运行的消费者
     */
    private static Map<String, ScheduledFuture> runningConsumerMap = new ConcurrentHashMap<>();


    /**
     * 开启消费者，
     * todo-保证消费成功？
     *
     * @param topic
     * @param millSecond
     * @param consumer
     * @return
     */
    public static boolean startConsumerMessage(String topic, long millSecond, Consumer<MessageInfo> consumer) {
        //向nameserver查询可用的broker
        List<BrokerInfo> ipList = BrokerStatusUtil.getAliveBrokerIpList().getBrokerInfoList();
        if (ipList.isEmpty()) {
            return false;
        }

        if (isConsumerAlreadyStarted(topic)) {
            return true;
        }

        ScheduledFuture future = scheduledExecutorService.scheduleAtFixedRate(() -> {
            String ip = ipList.get(NumberUtil.getRandom(ipList.size())).getIp();
            logger.info("consumer topic：" + topic + ", execute ip:" + ip);

            MessageInfo messageInfo = BrokerStatusUtil.consumeMessageFromBroker(ip, topic);
            if(messageInfo.isAllFinished()) {

                //todo 某个broker消费完成了怎么办？
                logger.warn("topic: " + topic + "broker: " + ip + " all finished!");
            }
            //执行消费逻辑
            consumer.accept(messageInfo);

            }, 1, millSecond, TimeUnit.MILLISECONDS
        );

        runningConsumerMap.put(topic, future);

        return true;
    }


    public static MessageInfo consumerOneMessage(String topic) {
        List<BrokerInfo> ipList = BrokerStatusUtil.getAliveBrokerIpList().getBrokerInfoList();
        if (ipList.isEmpty()) {
            return null;
        }
        String ip = ipList.get(NumberUtil.getRandom(ipList.size())).getIp();
        return BrokerStatusUtil.consumeMessageFromBroker(ip, topic);
    }

    /**
     * 消费者是否已存在， 且启动
     *
     * @param topic
     * @return
     */
    private static boolean isConsumerAlreadyStarted(String topic) {
        if(runningConsumerMap.containsKey(topic) && runningConsumerMap.get(topic).isCancelled()) {
            return true;
        }
        return false;
    }

    /**
     * 停止消费者
     *
     * @param topic
     * @return
     */
    public static boolean stopConsumerMessage(String topic) {
        ScheduledFuture future = runningConsumerMap.get(topic);
        if(Objects.nonNull(future)) {
            future.cancel(true);
        }
        return true;
    }
}
